package com.mango.ch03;

import java.io.IOException;
import java.util.SortedMap;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopN_CatDemoMapper extends Mapper<LongWritable, Text, NullWritable, Text> {
	// 定义本地top10需要的数据结构 sortedMap按照key自动排序
	private SortedMap<Double, String> top10cats = new TreeMap<Double, String>();
	private int N = 10;// 默认为top 10

	@Override
	protected void cleanup(Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		// 执行完map（）方法之后执行
		// 当每个映射器完整调用map方法完整在本地建立了topN列表后，将topN元素发出给ruducer
		for (String catAttr : top10cats.values()) {
			System.out.println("in mapper-->   key="+NullWritable.get()+"  value="+catAttr);
			context.write(NullWritable.get(), new Text(catAttr));
		}
	}

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, NullWritable, Text>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		String[] tokens = value.toString().split(",");
		// data format: 12,cat1,cat1
		Double weight = Double.valueOf(tokens[0]);
		this.top10cats.put(weight, value.toString());// 将体重作为键 ,整行数据作为值
		// 只保留N个元素
		// 删除体重最小的元素
		if (top10cats.size() > N)
			top10cats.remove(top10cats.firstKey());
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		// 执行map方法之前调用
		Configuration conf = context.getConfiguration();
		this.N = Integer.valueOf(conf.get("top.n"));// 获取配置文件中设置的N的限制数
	}

}
